package mr

import "C"
import (
	"fmt"
	"log"
	"sync"
	"time"
)
import "net"
import "os"
import "net/rpc"
import "net/http"

var coorLog = log.New(os.Stdout, "coordinator: ", log.LstdFlags)

// 任务超时时间（秒）
const TaskTimeout = 10

type Coordinator struct {
	//协作者的阶段
	Phase string
	//任务列表
	mapTasks    []*CoorTask
	reduceTasks []*CoorTask
	//map任务个数
	nMap int
	//reduce任务个数
	nReduce int
	//互斥锁
	mu sync.Mutex
	//已完成map任务个数
	completeMapCount int
	//已完成reduce任务个数
	completeReduceCount int

	//未分配的map任务队列
	mapTaskQueue []int
	//未分配的reduce任务队列
	reduceTaskQueue []int

	//处理中的任务队列，用以统计时间
	inProgressTasks map[int]*CoorTask

	//下一个任务Id
	nextTaskId int
	//退出主进程
	shouldExit bool

	//任务请求通道
	taskReqChan chan struct {
		req   *TaskReqArgs
		reply *TaskReplyArgs
		done  chan struct{}
	}
	//任务提交通道
	taskCompleteChan chan struct {
		req   *TaskCompleteReqArgs
		reply *TaskCompletedReplyArgs
		done  chan struct{}
	}
	//定时任务通道，空结构体仅用于表示信号
	//checkTimeoutChan chan struct{}
	//关闭通道
	shutdownChan chan struct{}
}

// Your code here -- RPC handlers for the worker to call.

//
// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
//
//func (c *Coordinator) Example(args *ExampleArgs, reply *ExampleReply) error {
//	c.mu.Lock()
//	defer c.mu.Unlock()
//	if args.X == -1 {
//		c.shouldExit = true
//	}
//	reply.Y = args.X + 1
//	return nil
//}

//检测是否存在超时任务
func (c *Coordinator) checkTimeOutTasks() {
	now := time.Now()
	for taskId, task := range c.inProgressTasks {
		//任务超时
		if now.After(task.Deadline) {
			var taskList *[]int
			if task.Type == MapTask {
				taskList = &c.mapTaskQueue
			} else {
				taskList = &c.reduceTaskQueue
			}
			//重置任务状态
			task.State = TaskUnassigned
			//重新放入未分配队列中
			*taskList = append(*taskList, task.TaskId)

			coorLog.Printf("任务 %d (%s) 超时，将重新分配\n", taskId, task.Type)
			//从处理中任务中清除
			delete(c.inProgressTasks, taskId)
		}
	}
}

func (c *Coordinator) startTimeoutChecker() {
	ticker := time.NewTicker(TaskTimeout * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			coorLog.Println("执行定时任务，检查是否存在过期任务")
			//向定时任务通道发信号，表示开始检查
			//c.checkTimeoutChan <- struct{}{}
		}
	}
}

//处理所有通道
func (c *Coordinator) processChannels() {
	ticker := time.NewTicker(TaskTimeout * time.Second)
	defer ticker.Stop()
	for {
		select {
		//处理请求任务
		case req := <-c.taskReqChan:
			//执行具体的处理逻辑
			c.ProcessTaskReq(req.req, req.reply)
			if req.done != nil {
				//等待处理完毕后发送结束信号
				req.done <- struct{}{}
			}
		//	处理提交任务
		case complete := <-c.taskCompleteChan:
			c.TaskComplete(complete.req, complete.reply)
			if complete.done != nil {
				complete.done <- struct{}{}
			}
		//定时器
		case <-ticker.C:
			coorLog.Println("执行定时任务，检查是否存在过期任务")
			//向定时任务通道发信号，表示开始检查
			//	检查超时任务
			c.checkTimeOutTasks()
		//	任务结束了，直接关闭
		case <-c.shutdownChan:
			coorLog.Println("收到关闭信号，退出主循环")
			return
		}
	}
}

// 初始化任务回复结构体
func (c *Coordinator) InitTaskReply(reply *TaskReplyArgs, taskPhase string, fileNames []string, taskID, nMap, nReduce int, outputFile string) error {

	// 初始化结构体字段
	reply.TaskPhase = taskPhase
	//reply.FileNames = make([]string, len(fileNames))
	//copy(reply.FileNames, fileNames)
	reply.FileNames = fileNames
	reply.TaskId = taskID
	reply.NMap = nMap
	reply.NReduce = nReduce
	reply.OutputFile = outputFile

	return nil
}

//针对请求新任务的处理流程
func (c *Coordinator) ProcessTaskReq(args *TaskReqArgs, reply *TaskReplyArgs) error {

	//检测是否存在超时任务并处理
	c.checkTimeOutTasks()

	if c.Phase == DonePhase {
		//已经处于Done阶段，直接拒绝请求
		_ = c.InitTaskReply(reply, c.Phase, nil, -1, c.nMap, c.nReduce, "")
		return nil
	}
	//当前是map阶段
	if c.Phase == MapPhase {
		//还有map任务未执行完，直接分配
		if len(c.mapTaskQueue) > 0 {
			//取出未完成任务的id
			index := c.mapTaskQueue[0]
			//未完成任务队列长度减一
			c.mapTaskQueue = c.mapTaskQueue[1:]
			//取出具体的任务
			task := c.mapTasks[index]
			//任务初始化
			task.State = TaskInProgress
			task.Deadline = time.Now().Add(TaskTimeout * time.Second)
			//任务放入执行队列中
			c.inProgressTasks[task.TaskId] = task
			//初始化文件列表
			files := make([]string, 1)
			files[0] = task.MapFile
			//回复参数初始化
			_ = c.InitTaskReply(reply, c.Phase, files, task.TaskId, c.nMap, c.nReduce, task.OutputFile)

			coorLog.Printf("分配map任务 %d\n", task.TaskId)
		} else {
			//map任务没有完成，但没空闲的任务，等待重试
			_ = c.InitTaskReply(reply, c.Phase, nil, -1, c.nMap, c.nReduce, "")
		}
		return nil
	}

	//处理reduce任务
	//有未分发的reduce任务
	if len(c.reduceTaskQueue) > 0 {
		index := c.reduceTaskQueue[0] - c.nMap
		c.reduceTaskQueue = c.reduceTaskQueue[1:]
		task := c.reduceTasks[index]
		//任务初始化
		task.State = TaskInProgress
		task.Deadline = time.Now().Add(TaskTimeout * time.Second)
		//任务放入执行队列中
		//index是需要，task id从c.nmap开始
		c.inProgressTasks[task.TaskId] = task
		//回复参数初始化
		_ = c.InitTaskReply(reply, c.Phase, task.Intermediates, task.TaskId, c.nMap, c.nReduce, task.OutputFile)

		coorLog.Printf("分配reduce任务 %d\n", task.TaskId)
		return nil
	} else { //reduce任务已经全部分配，但是尚未完成
		_ = c.InitTaskReply(reply, c.Phase, nil, -1, c.nMap, c.nReduce, "")
		return nil
	}
}

func (c *Coordinator) TaskComplete(args *TaskCompleteReqArgs, reply *TaskCompletedReplyArgs) error {

	// 阶段匹配检查（优先于状态检查），阶段不匹配，说明任务过期，直接跳过就好
	if (args.TaskType == MapTask && c.Phase != MapPhase) ||
		(args.TaskType == ReduceTask && c.Phase != ReducePhase) {
		coorLog.Fatalf("当前任务阶段%s不匹配任务: TaskID=%d, Type=%s\n",
			c.Phase, args.TaskId, args.TaskType)
		return nil // 阶段不匹配，直接忽略
	}

	var task *CoorTask

	//获取具体的任务
	if args.TaskType == MapTask {
		task = c.mapTasks[args.TaskId]
	} else if args.TaskType == ReduceTask {
		task = c.reduceTasks[args.TaskId-c.nMap]
	} else {
		coorLog.Fatalf("无效任务阶段: %s", args.TaskType)
		return nil
	}

	//当前任务已经处于结束状态
	if task.State == TaskFinished {
		coorLog.Fatalf("忽略重复的任务完成请求: TaskID=%d, Type=%s\n",
			args.TaskId, args.TaskType)
		return nil
	}
	//当前任务属于未完成任务，判断并修改状态
	//更新任务状态
	task.State = TaskFinished
	//从未完成任务队列中去除
	delete(c.inProgressTasks, task.TaskId)
	coorLog.Printf("任务完成: TaskID=%d, Type=%s\n", args.TaskId, args.TaskType)
	//如果是map任务，判断是否要进入reduce阶段
	if args.TaskType == MapTask {
		//完成数加一
		c.completeMapCount++
		//若全部任务完成就进入reduce阶段
		if c.completeMapCount == c.nMap {
			coorLog.Printf("进入reduce阶段")
			c.Phase = ReducePhase
		}
	} else {
		c.completeReduceCount++
		//若全部任务都已分配且无执行中任务，就进入Done阶段
		if c.completeReduceCount == c.nReduce && len(c.inProgressTasks) == 0 {

			coorLog.Printf("进入Done阶段，等待退出")
			//向关闭通道添加信息直接关闭通道拒绝服务
			c.shutdownChan <- struct{}{}
			c.Phase = DonePhase
			//标记当前应该退出
			c.mu.Lock()
			c.shouldExit = true
			c.mu.Unlock()
		}
	}
	return nil
}

// 实现RPC方法 - 请求任务
func (c *Coordinator) RequestTask(args *TaskReqArgs, reply *TaskReplyArgs) error {
	//减少死锁
	done := make(chan struct{})
	//发送信号开始执行任务
	c.taskReqChan <- struct {
		req   *TaskReqArgs
		reply *TaskReplyArgs
		done  chan struct{}
	}{args, reply, done}
	//等待任务执行完成
	<-done
	return nil
}

// 实现RPC方法 - 任务完成
func (c *Coordinator) CompleteTask(args *TaskCompleteReqArgs, reply *TaskCompletedReplyArgs) error {
	done := make(chan struct{})
	//发送信号开始执行任务
	c.taskCompleteChan <- struct {
		req   *TaskCompleteReqArgs
		reply *TaskCompletedReplyArgs
		done  chan struct{}
	}{args, reply, done}
	//等待任务执行完成
	<-done
	return nil
}

//
// start a thread that listens for RPCs from worker.go
//
// 启动协调器的RPC服务，监听Unix域套接字
func (c *Coordinator) server() {

	// 注册Coordinator类型的实例c作为RPC服务
	rpc.Register(c)

	// 将RPC服务绑定到HTTP协议，使用默认路径：
	rpc.HandleHTTP()

	// 生成唯一的Unix域套接字路径（如/var/tmp/824-mr-<uid>）
	sockname := coordinatorSock()

	// 启动前删除可能存在的旧套接字文件
	os.Remove(sockname)

	// 创建Unix域套接字并开始监听
	l, e := net.Listen("unix", sockname)
	if e != nil {
		coorLog.Fatal("listen error:", e)
	}
	coorLog.Println("套接字创建成功")

	//启动定时过期任务检查器,定时向通道发送信号
	//go c.startTimeoutChecker()

	//处理所有的通道信息
	go c.processChannels()

	// 在后台goroutine中启动HTTP服务
	go http.Serve(l, nil)
}

//
// main/mrcoordinator.go calls Done() periodically to find out
// if the entire job has finished.
//
func (c *Coordinator) Done() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.shouldExit
}

//
// create a Coordinator.
// main/mrcoordinator.go calls this function.
// nReduce is the number of reduce tasks to use.
//
// 在Coordinator中添加获取任务ID的方法
func (c *Coordinator) getNextTaskId() int {
	id := c.nextTaskId
	c.nextTaskId++
	return id
}
func MakeCoordinator(files []string, nReduce int) *Coordinator {
	c := Coordinator{
		Phase:               MapPhase,
		nMap:                len(files),
		nReduce:             nReduce,
		completeMapCount:    0,
		completeReduceCount: 0,
		inProgressTasks:     make(map[int]*CoorTask),
		nextTaskId:          0,
		shouldExit:          false,
		taskReqChan: make(chan struct {
			req   *TaskReqArgs
			reply *TaskReplyArgs
			done  chan struct{}
		}),
		taskCompleteChan: make(chan struct {
			req   *TaskCompleteReqArgs
			reply *TaskCompletedReplyArgs
			done  chan struct{}
		}),
		//checkTimeoutChan: make(chan struct{}),
		shutdownChan: make(chan struct{}, 1),
	}

	//创建map任务
	nMap := c.nMap
	c.mapTasks = make([]*CoorTask, nMap)
	c.mapTaskQueue = make([]int, nMap)
	for i, file := range files {
		task := CoorTask{
			Type:     MapTask,
			State:    TaskUnassigned,
			TaskId:   c.getNextTaskId(),
			MapFile:  file,
			Deadline: time.Time{},
		}
		//任务填充
		c.mapTasks[i] = &task
		//填充未完成的任务队列
		c.mapTaskQueue[i] = task.TaskId
	}

	//创建Reduce任务
	c.reduceTasks = make([]*CoorTask, nReduce)
	c.reduceTaskQueue = make([]int, nReduce)
	for j := range c.reduceTasks {
		//提前分配对应的中间文件
		intermediates := make([]string, nMap)
		for i := 0; i < nMap; i++ {
			intermediates[i] = fmt.Sprintf("mr-%d-%d", i, j)
		}
		task := CoorTask{
			Type:          ReduceTask,
			State:         TaskUnassigned,
			TaskId:        c.getNextTaskId(),
			Intermediates: intermediates,
			OutputFile:    fmt.Sprintf("mr-out-%d", j),
			Deadline:      time.Time{},
		}
		//任务填充
		c.reduceTasks[j] = &task
		//填充未完成的任务队列
		c.reduceTaskQueue[j] = task.TaskId
	}
	c.server()
	return &c
}
